package org.bing.engine.common.scheduler;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.bing.engine.common.logging.Log;
import org.bing.engine.common.logging.LogFactory;

public class SimpleScheduler {
    private static final Log logger = LogFactory.getLog(SimpleScheduler.class);
    private static final ScheduledExecutor executor = new ScheduledExecutor(1);
    private static final Map<String, SimpleTaskProxy> register = new ConcurrentHashMap<String, SimpleTaskProxy>();

    public void startup() {
        // TODO impl state manage, concurrent conflict
        logger.info("Startup simple scheduler ...");
        // register expired handle task
        long period = 5 * 1000;
        long delay = period - System.currentTimeMillis() % period;
        executor.scheduleAtFixedRate(new SimpleExpireTask(this, register), delay, period, TimeUnit.MILLISECONDS);
    }

    public synchronized void shutdown() {
        executor.shutdown();
        logger.info("Shutdown monitor's scheduler.");
    }

    @SuppressWarnings("rawtypes")
    public synchronized void schedule(String name, final Runnable task, Date beginDate, Date endDate, long period, boolean override) {
        if (register.containsKey(name)) {
            if (!override) {
                throw new RuntimeException("Task " + name + " is existed!");
            } else {
                this.cancel(name);
            }
        }
        long pms = period * 1000;
        long begin = beginDate.getTime();
        long end = endDate.getTime();
        long now = System.currentTimeMillis();
        long delay = 0;
        if (now > begin) {
            delay = pms - now % pms;
        } else {
            delay = begin - now;
        }
        Runnable proxy = null;
        if (!(task instanceof AbstractTask)) {
            proxy = new Runnable() {
                public void run() {
                    try {
                        if (logger.isDebugEnabled()) {
                            logger.debug(" - start task " + this.getClass().getSimpleName());
                        }
                        task.run();
                    } catch (Throwable e) {
                        logger.error("Fail to run task ", e);
                    }
                }
            };
        } else {
            proxy = task;
        }
        ScheduledFuture future = executor.scheduleAtFixedRate(proxy, delay, pms, TimeUnit.MILLISECONDS);
        register.put(name, new SimpleTaskProxy(name, begin, end, future));
        logger.info("Add task to scheduler " + name);
    }

    public void schedule(String name, final Runnable task, Date beginDate, long period) {
        this.schedule(name, task, beginDate, new Date(Long.MAX_VALUE), period, false);
    }

    public void schedule(String name, final Runnable task, long period) {
        this.schedule(name, task, new Date(), new Date(Long.MAX_VALUE), period, false);
    }

    public synchronized void cancel(String name) {
        SimpleTaskProxy proxy = register.get(name);
        if (proxy != null && proxy.getFuture() != null) {
            logger.info("Ready to cancle task: " + name);
            proxy.getFuture().cancel(false);
        }
        register.remove(name);
    }

    public synchronized boolean exists(String name) {
        return register.containsKey(name);
    }
}
